/*
 * Copyright 2011 LMAX Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package test.threads.queue.unstressed;

import com.lmax.disruptor.*;

/**
 * No batching!
 */
public final class NoBatchingEventProcessor<T>
        implements EventProcessor {
    private volatile boolean running = true;
    private ExceptionHandler exceptionHandler = new FatalExceptionHandler();
    private final RingBuffer<T> ringBuffer;
    private final SequenceBarrier sequenceBarrier;
    private final EventHandler<T> eventHandler;
    private final Sequence sequence = new Sequence( Sequencer.INITIAL_CURSOR_VALUE );

    /**
     * Construct a {@link com.lmax.disruptor.EventProcessor} that will automatically track the progress by updating its sequence when
     * the {@link com.lmax.disruptor.EventHandler#onEvent(Object, long, boolean)} method returns.
     *
     * @param ringBuffer      to which events are published.
     * @param sequenceBarrier on which it is waiting.
     * @param eventHandler    is the delegate to which events are dispatched.
     */
    public NoBatchingEventProcessor( final RingBuffer<T> ringBuffer,
                                     final SequenceBarrier sequenceBarrier,
                                     final EventHandler<T> eventHandler ) {
        this.ringBuffer = ringBuffer;
        this.sequenceBarrier = sequenceBarrier;
        this.eventHandler = eventHandler;
    }

    /**
     * Construct a batch event processor that will allow a {@link com.lmax.disruptor.SequenceReportingEventHandler}
     * to callback and update its sequence within a batch.  The Sequence will be updated at the end of
     * a batch regardless.
     *
     * @param ringBuffer      to which events are published.
     * @param sequenceBarrier on which it is waiting.
     * @param eventHandler    is the delegate to which events are dispatched.
     */
    public NoBatchingEventProcessor( final RingBuffer<T> ringBuffer,
                                     final SequenceBarrier sequenceBarrier,
                                     final SequenceReportingEventHandler<T> eventHandler ) {
        this.ringBuffer = ringBuffer;
        this.sequenceBarrier = sequenceBarrier;
        this.eventHandler = eventHandler;
        eventHandler.setSequenceCallback( sequence );
    }

    @Override
    public Sequence getSequence() {
        return sequence;
    }

    @Override
    public void halt() {
        running = false;
        sequenceBarrier.alert();
    }

    /**
     * Set a new {@link com.lmax.disruptor.ExceptionHandler} for handling exceptions propagated out of the {@link NoBatchingEventProcessor}
     *
     * @param exceptionHandler to replace the existing exceptionHandler.
     */
    public void setExceptionHandler( final ExceptionHandler exceptionHandler ) {
        if ( null == exceptionHandler ) {
            throw new NullPointerException();
        }

        this.exceptionHandler = exceptionHandler;
    }

    /**
     * Get the {@link com.lmax.disruptor.SequenceBarrier} the {@link com.lmax.disruptor.EventProcessor} is waiting on.
     *
     * @return the sequenceBarrier this {@link com.lmax.disruptor.EventProcessor} is using.
     */
    public SequenceBarrier getSequenceBarrier() {
        return sequenceBarrier;
    }

    /** It is ok to have another thread rerun this method after a halt(). */
    @Override
    public void run() {
        sequenceBarrier.clearAlert();
        running = true;

        T event = null;
        long nextSequence = sequence.get() + 1L;
        while ( true ) {
            try {
                final long availableSequence = sequenceBarrier.waitFor( nextSequence );
                if ( nextSequence <= availableSequence ) {
                    event = ringBuffer.get( nextSequence );
                    eventHandler.onEvent( event, nextSequence, nextSequence == availableSequence );
                    nextSequence++;
                    sequence.set( nextSequence - 1L );
                }

            } catch ( final AlertException ex ) {
                if ( !running ) {
                    break;
                }
            } catch ( final Exception ex ) {
                exceptionHandler.handleEventException( ex, nextSequence, event );
                sequence.set( nextSequence );
                nextSequence++;
            }
        }
    }
}